#include <iostream>
#include <cstring>

#include "TcpConnection.h"
#include "EventLoop.h"
#include "StringPiece.h"

TcpConnection::TcpConnection(EventLoop* loop, int connFd, 
            const InetAddress& serverAddr, const InetAddress& peerAddr) :
        loop_(loop),
        connSocket_(connFd),
        channel_(loop, connFd),
        serverAddr_(serverAddr),
        peerAddr_(peerAddr),
        state_(connecting)
    {
        channel_.setReadCallback(std::bind(&TcpConnection::handleRead, this));
        channel_.setWriteCallback(std::bind(&TcpConnection::handleWrite, this));
        channel_.setCloseCallback(std::bind(&TcpConnection::handleClose, this));
        connSocket_.setKeepAlive(true);

        // 不在构造函数就enableReading，因为此时还没有设置用户的messageCb
    }

TcpConnection::~TcpConnection() {

}

void TcpConnection::send(const char* str, size_t len) {
    loop_->runInLoop(std::bind(&TcpConnection::sendInLoop, this, str, len));
}
void TcpConnection::send(const StringPiece& str) {
    loop_->runInLoop(std::bind(&TcpConnection::sendInLoop, this, str.data(), str.size()));
}

void TcpConnection::sendInLoop(const char* str, size_t len) {
    loop_->assertInLoopThread();

    if (state_ == disconnected) {
        std::cout << "disconnected, give up writing" << std::endl;
        abort();
    }

    int remaining = len;
    
    if (!channel_.isWriting()) {
        int ret = connSocket_.write(str, len);
        if (ret >= 0) {
            remaining -= ret;
            if (ret == len && writeCompleteCallback_) {
                loop_->queueInLoop(std::bind(writeCompleteCallback_, shared_from_this()));
            }
            if (ret < len) {
                outputBuf_.append(str + ret, len - ret);
                channel_.enableWriting();
            }
        } else {
            if (errno != EWOULDBLOCK) {
                std::cout << "TcpConnection write error" << std::endl;
                return;
            }
            
        }
    }
    if (remaining > 0) {
        if (highWaterMarkCallback_ && outputBuf_.readableBytes() + remaining > highWaterMark_) {
            loop_->queueInLoop(std::bind(highWaterMarkCallback_, shared_from_this(), outputBuf_.readableBytes() + remaining));
        }
        outputBuf_.append(str + len - remaining, remaining);
        channel_.enableWriting();
    }
}

void TcpConnection::shutdown() {
    if (state_ == connected) {
        state_ = disconnecting;
        loop_->runInLoop(std::bind(&TcpConnection::shutdownInLoop, shared_from_this()));
    }
}

void TcpConnection::shutdownInLoop() {
    loop_->assertInLoopThread();
    if (!channel_.isWriting()) {
        connSocket_.shutdownWrite();
    }
}

void TcpConnection::forceClose() {
    loop_->runInLoop(std::bind(&TcpConnection::forceCloseInLoop, shared_from_this()));
}

void TcpConnection::forceCloseInLoop() {
    loop_->assertInLoopThread();
    handleClose();
}

void TcpConnection::connectEstablished() {
    loop_->assertInLoopThread();
    state_ = connected;
    channel_.enableReading();

    connectionCallback_(shared_from_this());
}

void TcpConnection::connectDestroyed() {
    std::cout << "TcpConnection::connectDestroyed" << std::endl;
    connectionCallback_(shared_from_this());
}

void TcpConnection::handleRead() {
    int savedErrno;
    int ret = inputBuf_.readFd(connSocket_.getFd(), &savedErrno);
    if (ret > 0) {
        messageCallback_(shared_from_this(), inputBuf_);
    } else if (ret == 0) {
        handleClose();
    } else {
        std::cout << "TcpConnection handleRead error" << std::endl;
    }
}

void TcpConnection::handleClose() {
    std::cout << "handleClose" << std::endl;  
    state_ = disconnected;
    channel_.disableAll();
    closeCallback_(shared_from_this());
}

void TcpConnection::handleWrite() {
    if (channel_.isWriting()) {
        int ret = connSocket_.write(outputBuf_.peek(), outputBuf_.readableBytes());
        if (ret > 0) {
            outputBuf_.retrive(ret);
            if (outputBuf_.readableBytes() == 0) {
                if (writeCompleteCallback_) {
                    loop_->queueInLoop(std::bind(writeCompleteCallback_, shared_from_this()));
                }
                channel_.disableWriting();
                if (state_ == disconnecting) {
                    shutdownInLoop();
                }
            }
        } else {
            std::cout << "handleWrite error" << std::endl;
            abort();
        }
    }
}
